package com.gxljc.bear.crawler.itaogao.tmtpost;

import com.mongodb.BasicDBObject;
import com.mongodb.Bytes;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.gxljc.commons.mongo.cli.BaseMongoRepository;
import com.gxljc.bear.crawler.base.BaseMongoTools;
import com.gxljc.bear.crawler.base.DolphinCrawlerConsts;
import com.gxljc.bear.crawler.base.DolphinFetchData;
import com.gxljc.bear.crawler.itaogao.ItaogaoConst;
import com.gxljc.bear.crawler.util.DateUtil;
import com.gxljc.bear.crawler.util.HtmlUtil;
import com.gxljc.bear.crawler.util.SparkUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;

import java.io.Serializable;
import java.lang.reflect.Field;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * 钛媒体 PAGE seed 数据爬虫。
 *
 * @author tanghaitao
 * @since 2022-8-5
 */
public class TmtpostPageSeedCrawler implements Serializable {
	public static Logger LOG = Logger.getLogger(TmtpostPageSeedCrawler.class);
	private Boolean proxy = false;

	public TmtpostPageSeedCrawler(Boolean proxy) {
		this.proxy = proxy;
	}

	public void crawl() throws Exception {
		List<String> seeds = getSeed();
		if (CollectionUtils.isEmpty(seeds)) {
			LOG.error("page is empty");
			return;
		}
		crawl(seeds);
	}

	public void crawlNoSpark(List<String> seeds) {
		for (String url : seeds) {
			try {
				crawl(url);
			} catch (Exception e) {
				e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
			}
		}
	}

	public void crawl(List<String> seeds) {
		int coreMax = 15;
		JavaSparkContext jsc = SparkUtil.createCommonsSparkContext(
				"dolphin-钛媒体page 种子爬取-" + seeds.size(), coreMax,
				coreMax * 2, TmtpostPageSeedCrawler.class);
		JavaRDD<String> seedsRDD = jsc.parallelize(new ArrayList<String>(seeds));
		long count = seedsRDD.mapPartitions(
				new FlatMapFunction<Iterator<String>, Integer>() {
					private static final long serialVersionUID = 1L;

					@Override
					public Iterable<Integer> call(Iterator<String> seeds)
							throws Exception {
						int successCnt = 0;
						while (seeds.hasNext()) {
							try {
								String seed = seeds.next();
								int ret = crawl(seed);
								if (ret == 1)
									successCnt++;
								if (successCnt % 100 == 0) {
									LOG.info("successCnt=" + successCnt);
								}
							} catch (Exception er) {
								er.printStackTrace();
							}
						}
						return Arrays.asList(successCnt);
					}
				}).count();
		LOG.info("all count=" + count);
		jsc.stop();
	}

	//单条url入口
	public int crawl(String url) throws Exception {
		LOG.info("crawl url = " + url);
		DolphinFetchData fetchData = DolphinFetchData.getInstance(DolphinCrawlerConsts.CrawlerChannelType.MOVIE.getName());
		byte[] htmlByte = fetchData.getHtml(url, proxy, ItaogaoConst.CRAWL_TIMEOUT);
		if (htmlByte == null) {
			LOG.error("page result is empty url = " + url);
			return DolphinCrawlerConsts.RESULT_NO;
		}
		String html = new String(htmlByte);
		if (StringUtils.isEmpty(html)) {
			LOG.error("page result is empty url = " + url);
			return DolphinCrawlerConsts.RESULT_NO;
		}
		List<TmtpostPageTable> tables = extract(html);
		saveMongodb(tables);
		return DolphinCrawlerConsts.RESULT_YES;
	}

	private List<TmtpostPageTable> extract(String html) {
		Document doc = Jsoup.parse(html);
		Elements elements = doc.select(".mod-article-list ul li");
		if (CollectionUtils.isEmpty(elements)) return null;
		List<TmtpostPageTable> tables = new LinkedList<>();
		for (Element element : elements) {
			TmtpostPageTable table = new TmtpostPageTable();
			extractDetail(table, element);
			tables.add(table);
		}
		return tables;
	}

	//抽取id
	private String extractId(String href) {
		final String patternUrl = "/(\\d+).html";
		Pattern p = Pattern.compile(patternUrl);
		Matcher m = p.matcher(href);
		if (m.find()) {
			return m.group(1);
		}
		return null;
	}

	//抽取logo url
	private void extractLogoUrl(TmtpostPageTable table, Element element) {
		Elements elements = element.select(".pic a img");
		if (CollectionUtils.isEmpty(elements)) return;
		String image = elements.get(0).attr("src");
		table.setLogoUrl(image);
	}

	//抽取tag
	private void extractTag(TmtpostPageTable table, Element element) {
		Elements elements = element.select(".cont .tag a");
		if (CollectionUtils.isEmpty(elements)) return;
		List<String> tags = new LinkedList<String>();
		for (Element tag : elements) {
			tags.add(tag.text());
		}
		table.setTagName(tags);
	}

	//抽取发布时间
	private void extractPublishDate(TmtpostPageTable table, Element element) {
		Elements elements = element.select(".cont .info .author");
		if (CollectionUtils.isEmpty(elements)) return;
		String dateStr = elements.get(0).ownText();
		if (StringUtils.isEmpty(dateStr)) return;
		Date publishDate = DateUtil.parseFormat(dateStr, "yyyy-MM-dd HH:mm");
		table.setPublishDate(publishDate);
	}

	//抽取详细信息
	private void extractDetail(TmtpostPageTable table, Element element) {
		Elements elements = element.select(".cont h3 a");
		if (CollectionUtils.isEmpty(elements)) return;
		String title = elements.get(0).text();
		String url = elements.get(0).attr("href");
		if (StringUtils.isEmpty(url)) return;
		String id = extractId(url);
		if (StringUtils.isEmpty(id)) return;
		table.setPageId(id);
		table.setCrawlUrl(HtmlUtil.pictureUrlPref(url, ItaogaoConst.TMTPOST_DOMAIN));
		table.setCrawlFlag(0);
		if (!StringUtils.isEmpty(title)) {
			table.setTitle(title.trim());
		}
		extractPublishDate(table, element);
		extractTag(table, element);
		extractLogoUrl(table, element);
	}


	//save data to mongodb
	public void saveMongodb(List<TmtpostPageTable> tables) throws Exception {
		if (CollectionUtils.isEmpty(tables)) return;
		for (TmtpostPageTable table : tables) {
			saveMongodb(table);
		}
	}

	//save data to mongodb
	public void saveMongodb(TmtpostPageTable table) throws Exception {
		Query query = new Query();
		query.addCriteria(Criteria.where(DolphinCrawlerConsts.MONGODB_DEFAULT_ID).is(table.pageId));
		Update update = new Update();
		Field[] fields = table.getClass().getDeclaredFields();
		for (int i = 0; i < fields.length; i++) {
			Object value = fields[i].get(table);
			if (value == null) continue;
			if (value instanceof List) {
				if (CollectionUtils.isEmpty((List) value)) continue;
			}
			if (value instanceof String) {
				if (StringUtils.isEmpty((String) value)) continue;
			}
			update.set(fields[i].getName(), value);
		}
		BaseMongoRepository repo = BaseMongoTools.getInstance(DolphinCrawlerConsts.MongoDBName.ITAOGAO.getName());
		repo.upsert(query, update, ItaogaoConst.MONGODB_TMTPOST_TABLE_PAGE);
	}

	//构造URL seed
	public List<String> getSeed() throws Exception {
		List<TmtpostTagTable> urlSeeds = getUrlSeed();
		final int ONE_PAGE = 10; //一页条数
		if (CollectionUtils.isEmpty(urlSeeds)) return null;
		List<String> urls = new LinkedList<String>();
		for (TmtpostTagTable seed : urlSeeds) {
			int pageSize = seed.getPageSize();
			String url = seed.getCrawlUrl();
			int max = (pageSize % ONE_PAGE == 0) ? pageSize / ONE_PAGE : pageSize / ONE_PAGE + 1 ;
			for (int i = 1; i <= max; i++) {
				String newUrl = String.format("%s/%d", url, i);
				urls.add(newUrl);
			}
		}
		return urls;
	}

	//获取数据库seed
	public List<TmtpostTagTable> getUrlSeed() throws Exception {
		List<TmtpostTagTable> seeds = new LinkedList<TmtpostTagTable>();
		BaseMongoRepository repo = BaseMongoTools.getInstance(DolphinCrawlerConsts.MongoDBName.ITAOGAO.getName());
		BasicDBObject keys = new BasicDBObject();
		keys.put(ItaogaoConst.SCHEMA_COLUMN_NAME.PAGE_SIZE.getValue(), 1);
		keys.put(ItaogaoConst.SCHEMA_COLUMN_NAME.CRAWL_URL.getValue(), 1);
		DBObject query = new BasicDBObject();
		query.put(ItaogaoConst.SCHEMA_COLUMN_NAME.PAGE_SIZE.getValue(), new BasicDBObject("$gt", 0));
		DBCursor cursor = repo.getCollection(ItaogaoConst.MONGODB_TMTPOST_TABLE_TAG)
				.find(query, keys)
				.addOption(Bytes.QUERYOPTION_NOTIMEOUT);
		List<DBObject> dataIterator = cursor.toArray();
		for (DBObject data : dataIterator) {
			try {
				Object pageSizeObj = data.get(ItaogaoConst.SCHEMA_COLUMN_NAME.PAGE_SIZE.getValue());
				Object urlObj = data.get(ItaogaoConst.SCHEMA_COLUMN_NAME.CRAWL_URL.getValue());
				TmtpostTagTable table = new TmtpostTagTable();
				table.setPageSize(Integer.parseInt(pageSizeObj.toString()));
				table.setCrawlUrl(urlObj.toString());
				seeds.add(table);
			} catch (Exception er) {
				er.printStackTrace();
			}
		}
		return seeds;
	}
}
